package com.atguigu.market_analysis

import java.util
import java.util.Map

import org.apache.flink.api.common.state.{MapState, MapStateDescriptor}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

/**
 * 市场营销商业指标统计分析
 * 统计App点击量、APP下载量(不分渠道)
 *
 * Project: UserBehaviorAnalysis
 * Package: com.atguigu.market_analysis
 * Version: 1.0
 *
 * Created by  WangJX  on 2019/12/12 15:55
 */
object AppMarketingStatistics {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val value = env.addSource(new MySource())
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[MarketingUserBehavior](Time.seconds(1)) {
        override def extractTimestamp(element: MarketingUserBehavior): Long = element.timestamp
      })
      //      .filter(_.behavior != "UNINSTALL")
      .keyBy(_.behavior)
      .timeWindow(Time.seconds(20), Time.seconds(5))
      .apply(new MyApply())
      .keyBy(_.windowEnd)
      .process(new MyProcess())


    value.print("output")

    env.execute("AppMarketingStatistics job")

  }

}

case class BehaviorTypesCount(
                               behaviorTypes: String,
                               windowEnd: Long,
                               count: Long
                             )

class MyApply() extends WindowFunction[MarketingUserBehavior, BehaviorTypesCount, String, TimeWindow] {
  override def apply(key: String, window: TimeWindow, input: Iterable[MarketingUserBehavior], out: Collector[BehaviorTypesCount]): Unit = {
    out.collect(BehaviorTypesCount(key, window.getEnd, input.size))
  }
}

class MyProcess() extends KeyedProcessFunction[Long, BehaviorTypesCount, String] {

  lazy val mapState: MapState[String, Long] = getRuntimeContext.getMapState(new MapStateDescriptor[String, Long]("map_state", classOf[String], classOf[Long]))

  override def processElement(value: BehaviorTypesCount, ctx: KeyedProcessFunction[Long, BehaviorTypesCount, String]#Context, out: Collector[String]): Unit = {
    var count: Long = mapState.get(value.behaviorTypes)
    if (count != null) {
      //更新数据
      count = count + value.count
      mapState.put(value.behaviorTypes, count)
    } else {
      mapState.put(value.behaviorTypes, value.count)
    }

    ctx.timerService().registerEventTimeTimer(value.windowEnd + 1)

  }

  override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Long, BehaviorTypesCount, String]#OnTimerContext, out: Collector[String]): Unit = {

    val sb: StringBuilder = new StringBuilder()

    sb.append("===============================\n")
        .append("关闭时间：").append(timestamp - 1).append("\n")

    val value = mapState.iterator()

    while (value.hasNext){
      val map: Map.Entry[String, Long] = value.next()
      sb.append("behavior:").append(map.getKey)
        .append(" count:").append(map.getValue)
        .append("\n")
    }

    mapState.clear()

    out.collect(sb.toString())
  }
}


